Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

src/event_gate_lambda.py (2)

102-104: Runtime type validation is still missing.

Using cast() satisfies the type checker but doesn't prevent non-RSA keys from entering TOKEN_PUBLIC_KEYS, which will cause runtime failures in jwt.decode. The previous review's concern remains valid: if the identity service serves an EC/DSA key during rotation, EventGate will accept it during initialization but fail unpredictably during token verification.

Replace the list comprehension with an explicit loop that validates each key's type at runtime, as suggested in the previous review.

---

106-108: Add TypeError to exception handling.

The exception handler should catch TypeError to handle the runtime type validation suggested in the previous comment, ensuring consistent error reporting for all key initialization failures.

Apply this diff:

-except (requests.RequestException, ValueError, KeyError, UnsupportedAlgorithm) as exc:
+except (requests.RequestException, ValueError, KeyError, UnsupportedAlgorithm, TypeError) as exc:
     logger.exception("Failed to fetch or deserialize token public key from %s", TOKEN_PUBLIC_KEYS_URL)
     raise RuntimeError("Token public key initialization failed") from exc

src/event_gate_lambda.py (1)

225-237: Consider adding debug logging for key rotation observability.

The multi-key verification logic is correct and implements the expected behavior. However, for operational observability during key rotations, consider logging when verification fails with one key and falls back to another.

Apply this diff to add debug logging:

 def decode_jwt_all(token_encoded: str) -> Dict[str, Any]:
     """Decode JWT using any of the loaded public keys.
 
     Args:
         token_encoded: Encoded bearer JWT token string.
     """
-    for public_key in TOKEN_PUBLIC_KEYS:
+    for idx, public_key in enumerate(TOKEN_PUBLIC_KEYS):
         try:
             return jwt.decode(token_encoded, public_key, algorithms=["RS256"])
         except jwt.PyJWTError:
+            logger.debug("Token verification failed with public key %d, trying next", idx)
             continue
     raise jwt.PyJWTError("Verification failed for all public keys")

This would help operators confirm that key rotation is working as expected when monitoring logs.

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

tests/handlers/test_handler_token.py (1)

85-91: Consider adding a test for the Authorization: Bearer <token> flow.

The current tests cover the direct Bearer header and empty headers, but the standard Authorization header with Bearer scheme (the more common pattern) isn't explicitly tested here.

def test_extract_token_authorization_header():
    token = HandlerToken.extract_token({"Authorization": "Bearer mytoken123"})
    assert token == "mytoken123"


def test_extract_token_authorization_header_case_insensitive():
    token = HandlerToken.extract_token({"authorization": "bearer MYTOKEN"})
    assert token == "MYTOKEN"

src/handlers/handler_token.py (2)

44-47: Add validation for missing public_keys_url.

If neither TOKEN_PUBLIC_KEYS_URL nor TOKEN_PUBLIC_KEY_URL is present in the config, public_keys_url will be None. This will cause load_public_keys to fail with an unclear error when requests.get(None, ...) is called.

Consider validating during initialization:

     def __init__(self, config):
         self.provider_url: str = config.get(TOKEN_PROVIDER_URL, "")
         self.public_keys_url: str = config.get(TOKEN_PUBLIC_KEYS_URL) or config.get(TOKEN_PUBLIC_KEY_URL)
+        if not self.public_keys_url:
+            raise ValueError("Missing required config: token_public_keys_url or token_public_key_url")
         self.public_keys: list[RSAPublicKey] = []

---

94-100: Consider differentiating ExpiredSignatureError for faster failure.

Currently, all PyJWTError exceptions (including ExpiredSignatureError) cause the loop to try the next key. However, if a token is expired, it will be expired regardless of which key is used. You could short-circuit on ExpiredSignatureError to avoid unnecessary iterations.

         for public_key in self.public_keys:
             try:
                 return jwt.decode(token_encoded, public_key, algorithms=["RS256"])
+            except jwt.ExpiredSignatureError:
+                raise  # Expired tokens won't validate with any key
             except jwt.PyJWTError:
                 continue
         raise jwt.PyJWTError("Verification failed for all public keys")

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

src/handlers/handler_token.py (1)

81-82: Security: SSL verification disabled (already flagged).

This concern was raised in a previous review. Using verify=False when fetching cryptographic keys creates MITM vulnerability.

tests/handlers/test_handler_token.py (2)

70-73: Remove unused noqa directives.

The noqa: D401 directives on lines 70 and 85 are unused since this rule is not enabled. The unused parameters (token, algorithms, key) in mock stubs are intentional to match the jwt.decode signature.

-    def decode_side_effect(token, key, algorithms):  # noqa: D401 - test stub
+    def decode_side_effect(token, key, algorithms):

-    def always_fail(token, key, algorithms):  # noqa: D401 - test stub
+    def always_fail(token, key, algorithms):

---

94-100: Consider adding test for Authorization header extraction.

The current tests cover empty headers and direct Bearer header. Consider adding coverage for the Authorization: Bearer <token> format which is also supported by extract_token.

def test_extract_token_authorization_header():
    token = HandlerToken.extract_token({"Authorization": "Bearer mytoken123"})
    assert token == "mytoken123"

src/handlers/handler_token.py (2)

47-51: Consider validating public_keys_url is not None.

If neither TOKEN_PUBLIC_KEYS_URL nor TOKEN_PUBLIC_KEY_URL are present in config, public_keys_url will be None, causing requests.get(None, ...) to fail with a confusing error in load_public_keys.

     def __init__(self, config):
         self.provider_url: str = config.get(TOKEN_PROVIDER_URL, "")
-        self.public_keys_url: str = config.get(TOKEN_PUBLIC_KEYS_URL) or config.get(TOKEN_PUBLIC_KEY_URL)
+        self.public_keys_url: str = config.get(TOKEN_PUBLIC_KEYS_URL) or config.get(TOKEN_PUBLIC_KEY_URL) or ""
         self.public_keys: list[RSAPublicKey] = []
         self._last_loaded_at: datetime | None = None

Alternatively, raise a ValueError in __init__ if public_keys_url is falsy to fail fast with a clear error message.

---

96-98: Unsafe cast to RSAPublicKey may fail at runtime.

serialization.load_der_public_key() can return different key types (RSA, EC, Ed25519, etc.). If the key is not RSA, the cast will succeed but subsequent usage with jwt.decode (expecting RSA for RS256) may fail with an unclear error.

-            self.public_keys = [
-                cast(RSAPublicKey, serialization.load_der_public_key(base64.b64decode(raw_key))) for raw_key in raw_keys
-            ]
+            loaded_keys = []
+            for raw_key in raw_keys:
+                key = serialization.load_der_public_key(base64.b64decode(raw_key))
+                if not isinstance(key, RSAPublicKey):
+                    raise ValueError(f"Expected RSA public key, got {type(key).__name__}")
+                loaded_keys.append(key)
+            self.public_keys = loaded_keys

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

tests/handlers/test_handler_token.py (1)

104-121: Docstrings correctly reference 28 minutes.

The docstrings now accurately reflect the _REFRESH_INTERVAL of 28 minutes, addressing the previous review feedback.

src/handlers/handler_token.py (2)

79-83: Do not hard-code verify=False when fetching public keys.

requests.get(..., verify=False, ...) disables TLS certificate verification and opens a serious MITM risk, especially dangerous when downloading public keys used for JWT verification. This concern was already raised previously and still applies.

Consider making TLS verification configurable but default-on, and logging when it is disabled, e.g.:

@@
-logger = logging.getLogger(__name__)
-log_level = os.environ.get("LOG_LEVEL", "INFO")
-logger.setLevel(log_level)
+logger = logging.getLogger(__name__)
+log_level = os.environ.get("LOG_LEVEL", "INFO")
+logger.setLevel(log_level)
+
+# Controls SSL certificate verification when fetching token public keys.
+# Default: verify certificates. Set TOKEN_PUBLIC_KEYS_VERIFY_TLS=false to disable (e.g. in dev).
+_verify_env = os.environ.get("TOKEN_PUBLIC_KEYS_VERIFY_TLS", "true").lower()
+TOKEN_PUBLIC_KEYS_VERIFY_TLS = _verify_env not in {"0", "false", "no"}
@@
-        try:
-            response_json = requests.get(self.public_keys_url, verify=False, timeout=5).json()
+        try:
+            if not TOKEN_PUBLIC_KEYS_VERIFY_TLS:
+                logger.warning(
+                    "SSL certificate verification is disabled for token public keys URL %s",
+                    self.public_keys_url,
+                )
+            response_json = requests.get(
+                self.public_keys_url,
+                verify=TOKEN_PUBLIC_KEYS_VERIFY_TLS,
+                timeout=5,
+            ).json()

If your config system already has a place for this flag, you can wire it through there instead of using an env var, but the key points are: default to verify=True, make disabling explicit, and never leave verify=False hard-coded.

---

119-125: Narrow JWT exception handling to signature mismatches only.

Catching all jwt.PyJWTError means expired tokens, malformed tokens, etc. will unnecessarily try every key and then surface as a generic "Verification failed for all public keys", masking the real cause. This was called out in a prior review and is still present.

A pattern that keeps retries only for signature mismatches:

-        logger.debug("Decoding JWT")
-        for public_key in self.public_keys:
-            try:
-                return jwt.decode(token_encoded, public_key, algorithms=["RS256"])
-            except jwt.PyJWTError:
-                continue
-        raise jwt.PyJWTError("Verification failed for all public keys")
+        logger.debug("Decoding JWT")
+        last_signature_error: jwt.PyJWTError | None = None
+
+        for public_key in self.public_keys:
+            try:
+                return jwt.decode(token_encoded, public_key, algorithms=["RS256"])
+            except jwt.PyJWTError as exc:
+                # Only retry on signature mismatch; propagate other JWT errors immediately.
+                if isinstance(exc, jwt.exceptions.InvalidSignatureError):
+                    last_signature_error = exc
+                    continue
+                raise
+
+        if last_signature_error is not None:
+            # All keys tried, but signatures didn't match.
+            raise last_signature_error
+        raise jwt.PyJWTError("Verification failed: no public keys succeeded")

Please verify against the PyJWT version you use that jwt.exceptions.InvalidSignatureError is the correct class to distinguish signature mismatches.

tests/conftest.py (2)

61-61: Remove unused noqa directive.

Per static analysis, the D401 rule is not enabled, making the # noqa: D401 comment unnecessary.

-            def flush(self):  # noqa: D401 - simple stub
+            def flush(self):

---

97-101: Remove unused noqa directives.

The # noqa: D401 comments on lines 97 and 101 are unnecessary since D401 is not enabled.

-        def Object(self, _key):  # noqa: D401 - simple proxy
+        def Object(self, _key):
             return MockS3Object()

     class MockS3Resource:
-        def Bucket(self, _name):  # noqa: D401 - simple proxy
+        def Bucket(self, _name):
             return MockS3Bucket()

tests/handlers/test_handler_token.py (2)

70-70: Remove unused noqa directive.

The # noqa: D401 comment is unnecessary since D401 is not enabled.

-    def decode_side_effect(token, key, algorithms):  # noqa: D401 - test stub
+    def decode_side_effect(token, key, algorithms):

---

85-85: Remove unused noqa directive.

-    def always_fail(token, key, algorithms):  # noqa: D401 - test stub
+    def always_fail(token, key, algorithms):

src/handlers/handler_token.py (2)

47-52: Fail fast on missing public keys URL in config.

If both TOKEN_PUBLIC_KEYS_URL_KEY and TOKEN_PUBLIC_KEY_URL_KEY are absent, self.public_keys_url becomes None, and load_public_keys will later fail via requests.get(None, ...) with a less explicit error.

Consider validating this in __init__ and raising a clear configuration error early, e.g.:

     def __init__(self, config):
         self.provider_url: str = config.get(TOKEN_PROVIDER_URL_KEY, "")
-        self.public_keys_url: str = config.get(TOKEN_PUBLIC_KEYS_URL_KEY) or config.get(TOKEN_PUBLIC_KEY_URL_KEY)
+        self.public_keys_url: str | None = (
+            config.get(TOKEN_PUBLIC_KEYS_URL_KEY) or config.get(TOKEN_PUBLIC_KEY_URL_KEY)
+        )
         self.public_keys: list[RSAPublicKey] = []
         self._last_loaded_at: datetime | None = None
+
+        if not self.public_keys_url:
+            raise ValueError("Token public keys URL must be configured")

This makes misconfiguration easier to diagnose and avoids opaque networking errors.

---

53-70: Clarify behavior when no public keys have ever been loaded.

If _last_loaded_at is None, _refresh_keys_if_needed returns without loading, and decode_jwt then iterates over an empty self.public_keys and raises a generic PyJWTError. That makes first-use behavior dependent on external callers remembering to call load_public_keys().

Either lazily load on first use or fail with a more explicit error. For example:

     def _refresh_keys_if_needed(self) -> None:
@@
-        if self._last_loaded_at is None:
-            return
+        if self._last_loaded_at is None:
+            logger.debug("Token public keys not loaded yet, loading now")
+            try:
+                self.load_public_keys()
+            except RuntimeError:
+                logger.warning("Initial token public key load failed")
+            return

and/or in decode_jwt guard against an empty key set:

     def decode_jwt(self, token_encoded: str) -> Dict[str, Any]:
         self._refresh_keys_if_needed()
 
         logger.debug("Decoding JWT")
+        if not self.public_keys:
+            raise jwt.PyJWTError("No token public keys loaded for verification")

This makes operational issues much easier to debug.

Also applies to: 117-125

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro

tests/conftest.py (1)

33-122: Consider test isolation implications of module-scoped fixture.

The event_gate_module fixture uses scope="module", meaning it's shared across all tests in a test module. While this improves performance by reducing setup overhead, it can lead to test coupling if tests modify the imported module's state or its mocked dependencies. Ensure that tests using this fixture are stateless or properly reset any modified state.

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro